package com.cloudpark.mqtt;

import org.apache.log4j.Logger;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttSecurityException;
import org.eclipse.paho.client.mqttv3.MqttTopic;

import com.cloudpark.common.util.Constants;

/**
 * MQTT Publisher
 */

public class MQTTPub {
	
	private static final Logger log = Logger.getLogger(MQTTPub.class);
	
	private static MQTTPub mqttPub = null;
	private static MqttClient client = null;
	
	public static MQTTPub getSingleton() {
		if (null == mqttPub) {
			mqttPub = new MQTTPub();
		}
		return mqttPub;
	}
	
	public MqttClient getMqttClient() {
		if (null == client) {
			try {
				client = new MqttClient(Constants.MQTT_ADDRESS, Constants.MQTT_PUBLISHER);
			} catch (MqttException e) {
				log.error("new MqttClient error", e);
			}
			if (!client.isConnected()) {
				try {
					client.connect();
				} catch (MqttSecurityException e) {
					log.error("MqttClient connect error", e);
				} catch (MqttException e) {
					log.error("MqttClient connect error", e);
				}
			}
		}
		return client;
	}
	
	public MqttTopic getMqttTopic() {
		return getMqttClient().getTopic(Constants.MQTT_TOPIC_SERVER);
	}
	
	public boolean publish(String message) {
		try {
			log.info("mqtt sendMessage = " + message);
			MqttMessage _m = new MqttMessage(message.getBytes("utf8"));
	        _m.setQos(Constants.MQTT_QoS);
	        MqttDeliveryToken token = getMqttTopic().publish(_m);
	        token.waitForCompletion(Constants.MQTT_SLEEPTIMEOUT);
	        log.info("Delivery token " + token.hashCode() + " has been received : " + token.isComplete());
	        return token.isComplete();
		} catch (Exception e) {
			log.error("publish mqtt message error", e);
		}
        return false;
	}
}
